package com.ksyun.campus.client.util;



import com.ksyun.campus.client.domain.DataServerInfo;
import com.ksyun.campus.client.domain.MetaServerInfo;
import lombok.Data;
import lombok.Getter;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

@Component
@Getter
public class ZkUtil {

    private String connectString = "192.168.10.130:2181"; // 连接字符串

    private CuratorFramework client;


    private List<MetaServerInfo> metaServers = new ArrayList<>();
    private List<DataServerInfo> dataServers = new ArrayList<>();

    public ZkUtil() throws Exception {
        // 构造方法初始化Curator客户端
        init();
    }

    @PostConstruct
    public void init() throws Exception {
        // 初始化，与ZooKeeper建立连接，注册监听路径，当配置有变化时随时更新
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(500, 5);

        client = CuratorFrameworkFactory.builder().connectString(connectString)
                .sessionTimeoutMs(15 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .build();

        System.out.println("开始启动client");
        long startTime = System.currentTimeMillis();
        client.start(); // 很耗时间
        System.out.println("启动成功client");
        long endTime = System.currentTimeMillis();
        long duration = (endTime - startTime) / 1000;
        System.out.println("启动耗时: " + duration + "秒");

        // 初始化时获取当前节点数据
        refreshServerData();

        // 监听/server路径下的所有子节点变化
        PathChildrenCache cache = new PathChildrenCache(client, "/server", true);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                    case CHILD_REMOVED:
                    case CHILD_UPDATED:
                        refreshServerData();
                        break;
                    default:
                        break;
                }
            }
        });
        cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
    }

    // 刷新数据服务器和元数据服务器信息
    private void refreshServerData() throws Exception {
        metaServers.clear();
        dataServers.clear();

        Stat metaServerStat = client.checkExists().forPath("/server/metaServer");
        Stat dataServerStat = client.checkExists().forPath("/server/dataServer");

        if (metaServerStat != null) {
            List<String> servers = client.getChildren().forPath("/server/metaServer");
            for (String server : servers) {
                byte[] bytes = client.getData().forPath("/server/metaServer/" + server);
                String s= new String(bytes, java.nio.charset.StandardCharsets.US_ASCII);
                MetaServerInfo metaServerInfo = stringToMetaServerInfo(s);
                metaServers.add(metaServerInfo);
            }
        }

        if (dataServerStat != null) {
            List<String> servers = client.getChildren().forPath("/server/dataServer");
            for (String server : servers) {
                byte[] bytes = client.getData().forPath("/server/dataServer/" + server);
                String s= new String(bytes, java.nio.charset.StandardCharsets.US_ASCII);
                DataServerInfo dataServerInfo = stringToDataServerInfo(s);
                dataServers.add(dataServerInfo);
            }
        }

        System.out.println("metaServers:" + metaServers);
        System.out.println("dataServers:" + dataServers);
    }

    public MetaServerInfo stringToMetaServerInfo(String data) {
        String[] parts = data.split(",");

        String ipAddress = parts[0];
        int port = Integer.parseInt(parts[1]);

        MetaServerInfo metaServerInfo = new MetaServerInfo();
        metaServerInfo.setIpAddress(ipAddress);
        metaServerInfo.setPort(port);

        return metaServerInfo;
    }

    public DataServerInfo stringToDataServerInfo(String data) {
        String[] parts = data.split(",");

        DataServerInfo dataServerInfo = new DataServerInfo();

        dataServerInfo.setIpAddress(parts[0]);
        dataServerInfo.setPort(Integer.parseInt(parts[1]));
        dataServerInfo.setAzname(parts[2]);
        dataServerInfo.setRack(parts[3]);
        dataServerInfo.setZone(parts[4]);

        dataServerInfo.setCapacity(Integer.parseInt(parts[5]));
        dataServerInfo.setUseCapacity(Integer.parseInt(parts[6]));
        dataServerInfo.setFileTotal(Integer.parseInt(parts[7]));

        return dataServerInfo;
    }

    //TODO 获得一个MetaServer节点
    public String getMetaServer() {
        return "http://localhost:8000";
    }
}
